package raft

import (
	"fmt"
	"math/rand"
	"net/rpc"
	"strconv"
	"sync"
	"time"
)

const (
	STATE_FOLLOWER  = 1
	STATE_CONDIDATE = 2
	STATE_LEADER    = 3
)

//节点
type Node struct {
	Addr string
	Id   int
}

//节点注册响应
type NodeRegisterReply struct {
	//注册结果 true:成功
	Result bool
	//已经注册了的节点列表
	RegisterNodeSlc []Node
}

//raft
type Raft struct {
	sync.Mutex
	allNodeSlc []Node
	selfNodeId int
	selfNode   Node
	state      int
	//是否leader
	isLeader bool
	//重置选举超时的通道
	resetTimerChn chan struct{}
	//选举超时定时器
	electionTimer *time.Timer
	//选举超时时间
	electionTimeout time.Duration
	//心跳定时器
	heartbeatTimer *time.Timer
	//心跳超时时间
	heartBeatTimeout time.Duration
	//心跳停止通道
	heartbeatStopChn chan struct{}
	//当前节点的任期号
	Term int
	//当前节点投票给谁--节点id
	VotedFor int
	//日志切片
	LogSlc []LogEntry
	//当前节点最大日志索引值
	logIndex int
	//已知最大的已经提交的日志条目索引值
	commitIndex int
	//最后被引用到状态机的日志条目索引值
	applyIndex int
	//对于每一个服务器，需要发送给他的下一个日志条目的索引值（初始化为领导人最后索引值加1）
	nodeIdKeyNextIndexMap map[int]int
	//已经复制了的日志索引
	nodeIdKeyCopiedIndexMap map[int]int
}

//候选人投票
type CandidateVote struct {
	//候选人任期号
	Term int
	//候选人节点id
	NodeId int
	//候选人最后的日志索引
	LastLogIdx int
	//候选人最后的日志任期号
	LastLogTerm int
}

//投票响应
type VoteReply struct {
	//节点任期号
	NodeTerm int
	//true：投票给候选人
	Flag bool
}

//当前节点的全局raft指针变量
var GlobalRaftPtr *Raft

//各个节点RPC连接对象Map
var NodeIdKeyConnMap = make(map[int]*rpc.Client)

//NodeIdKeyConnMap锁
var NodeIdKeyConnMapLock = sync.Mutex{}

/*
 初始化raft
*/
func BuildRaft(allNodeSlc []Node, selfNode Node) *Raft {
	raft := &Raft{}
	raft.allNodeSlc = allNodeSlc
	raft.selfNodeId = selfNode.Id
	raft.selfNode = selfNode
	raft.state = STATE_FOLLOWER
	raft.isLeader = false
	raft.VotedFor = -1
	raft.resetTimerChn = make(chan struct{})
	//选举超时时间400~800ms
	rand.Seed(time.Now().UnixNano())
	timeoutRandNum := time.Duration(int64(rand.Intn(100)) + time.Now().UnixNano()%int64(400) + int64(raft.selfNodeId*10))
	raft.electionTimeout = time.Millisecond * (400 + timeoutRandNum)
	raft.electionTimer = time.NewTimer(raft.electionTimeout)
	//心跳时间 200ms
	raft.heartBeatTimeout = time.Millisecond * 200
	raft.heartbeatStopChn = make(chan struct{})
	raft.nodeIdKeyNextIndexMap = make(map[int]int, len(allNodeSlc))
	raft.nodeIdKeyCopiedIndexMap = make(map[int]int, len(allNodeSlc))
	//raft.LogSlc=make([]LogEntry,0)
	return raft
}

/**
选举守护
*/
func (this *Raft) ElectionDaemon() {
	fmt.Println(strconv.Itoa(time.Now().Nanosecond()) + "===>选举ElectionDaemon--等待超时...超时时间：" + this.electionTimeout.String())
	for {
		select {
		case <-this.resetTimerChn:
			//重置超时定时器
			//fmt.Println("通过resetTimerChn重置超时")
			this.electionTimer.Reset(this.electionTimeout)
		case <-this.electionTimer.C:
			//选举超时--没有在指定的时间内收到leader的心跳，当前节点编程candidate并发起选举投票
			go this.StartElection()
			fmt.Println(strconv.Itoa(time.Now().Nanosecond()) + "===>选举超时，发起选举")
			this.electionTimer.Reset(this.electionTimeout)
		}
		//fmt.Println("是否一直在for中")
	}
}

/**
发起投票
*/
func (this *Raft) StartElection() {
	fmt.Println("开始选举 StartElection...")
	//重置定时器
	this.resetTimerChn <- struct{}{}
	this.state = STATE_CONDIDATE
	//投票给自己
	this.VotedFor = this.selfNodeId
	candidateVoteStu := this.buildVoteArgs()
	var voteAgreeNum = 1
	var voteReplyChn = make(chan *VoteReply, len(this.allNodeSlc))
	var wg sync.WaitGroup
	for _, remoteNode := range this.allNodeSlc {
		if this.selfNodeId != remoteNode.Id {
			wg.Add(1)
			//向其他节点发起投票
			go func(remoteNode Node) {
				fmt.Println("向" + remoteNode.Addr + "发起投票")
				voteReplyStuPtr := &VoteReply{}
				Call(remoteNode, "Raft.VoteRpcCall", candidateVoteStu, voteReplyStuPtr)
				voteReplyChn <- voteReplyStuPtr
				wg.Done()
				//fmt.Println(remoteNode.Addr+"投票完成,结果:"+strconv.FormatBool(voteReplyStuPtr.Flag))
			}(remoteNode)
		}
	}
	fmt.Println("等待投票中。。。。")
	wg.Wait()
	close(voteReplyChn)
	for voteReplyPtr := range voteReplyChn {
		if voteReplyPtr.Flag {
			voteAgreeNum++
		}
	}
	fmt.Println("得票总数：" + strconv.Itoa(voteAgreeNum))
	//超过一半的节点同意
	if voteAgreeNum > (len(this.allNodeSlc)/2 + 1) {
		this.isLeader = true
		this.state = STATE_LEADER
		this.Term++
		//重置定时器
		this.resetTimerChn <- struct{}{}
		//初始化其他节点的nextIndex值
		this.resetAllNodeNextIndex()
		//发送日志一致性校验
		go this.SendLogConsistencyCheck()
		//发送心跳
		go this.SendHeartbeat()
	} else {
		//选举失败--停止当前的心跳发送
		this.heartbeatStopChn <- struct{}{}
	}
}

/**
Leader心跳发送
*/
func (this *Raft) SendHeartbeat() {
	if this.heartbeatTimer == nil {
		fmt.Println("创建心跳定时器")
		this.heartbeatTimer = time.NewTimer(this.heartBeatTimeout)
	} else {
		fmt.Println("已经存在心跳定时器，重置定时--开始心跳发送")
		this.heartbeatTimer.Reset(this.heartBeatTimeout)
	}
	var exitForFlag = false
	for {
		select {
		case <-this.heartbeatTimer.C:
			//重置节点的定时器
			this.resetTimerChn <- struct{}{}
			//重置心跳定时器
			this.heartbeatTimer.Reset(this.heartBeatTimeout)
			for _, remoteNode := range this.allNodeSlc {
				if remoteNode.Id != this.selfNodeId {
					go func(remoteNode Node) {
						preLogIndex := this.nodeIdKeyNextIndexMap[remoteNode.Id] - 1
						if preLogIndex < 0 {
							preLogIndex = 0
						}
						preLogTerm := 0
						var replicateLogSlc []LogEntry
						if len(this.LogSlc) > 0 {
							preLogTerm = this.LogSlc[preLogIndex].Term
							replicateLogSlc = this.LogSlc[this.nodeIdKeyNextIndexMap[remoteNode.Id]:]
						}
						logReplicate := LogReplicate{
							PreLogIndex:       preLogIndex,
							PreLogTerm:        preLogTerm,
							LogSlc:            replicateLogSlc,
							LeaderNode:        this.selfNode,
							LeaderCommitIndex: 0,
						}
						logReplicateReplyPtr := &LogReplicateReply{}
						fmt.Printf("向%v发送心跳和日志复制，心跳参数%v\n", remoteNode.Addr, logReplicate)
						Call(remoteNode, "Raft.HeartbeatRpcCall", logReplicate, logReplicateReplyPtr)
						this.Lock()
						if logReplicateReplyPtr.Result {
							this.nodeIdKeyNextIndexMap[remoteNode.Id] = len(this.LogSlc)
							this.nodeIdKeyCopiedIndexMap[remoteNode.Id] = this.logIndex
						} else if logReplicateReplyPtr.ConflictLogIdx != -1 {
							//找到冲突点,日志的复制从冲突点的下一个位置开始复制
							this.nodeIdKeyNextIndexMap[remoteNode.Id] = logReplicateReplyPtr.ConflictLogIdx + 1
							this.nodeIdKeyCopiedIndexMap[remoteNode.Id] = logReplicateReplyPtr.ConflictLogIdx
						} else {
							//找不到冲突的索引继续往下-1
							this.nodeIdKeyNextIndexMap[remoteNode.Id]--
							this.nodeIdKeyCopiedIndexMap[remoteNode.Id] = 0
						}
						if this.nodeIdKeyNextIndexMap[remoteNode.Id] < 0 {
							this.nodeIdKeyNextIndexMap[remoteNode.Id] = 0
						}
						fmt.Printf("%s返回的日志复制结果：%p,,,,,节点的nextId为：%v\n", remoteNode.Addr, logReplicateReplyPtr, this.nodeIdKeyNextIndexMap[remoteNode.Id])
						this.Unlock()
					}(remoteNode)
				}
			}
		case <-this.heartbeatStopChn:
			this.heartbeatTimer.Stop()
			exitForFlag = true
		}
		if exitForFlag {
			//停止心跳发送--当前节点失去leader
			fmt.Println(strconv.Itoa(this.selfNode.Id) + "失去Leader")
			break
		}
	}
}

/*
构建投票请求参数
*/
func (this *Raft) buildVoteArgs() CandidateVote {
	var candidateVote = CandidateVote{
		Term:   this.Term,
		NodeId: this.selfNodeId,
	}
	if len(this.LogSlc) == 0 {
		candidateVote.LastLogIdx = 0
		candidateVote.LastLogTerm = 0
	} else {
		candidateVote.LastLogIdx = len(this.LogSlc)
		candidateVote.LastLogTerm = this.LogSlc[this.commitIndex].Term
	}
	return candidateVote
}
